/**
 * Copyright (C) 2013 SHANGHAI VOLKSWAGEN, All rights reserved.
 * License version 1.0, a copy of which has been included with this.
 *
 * @File name：com.market.cart.server.consumer.CartConsumer
 * @Create on：2024/3/6
 * @Author：liuchang
 */
package com.market.cart.server.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.market.cart.common.domain.CartInfo;
import com.market.cart.server.config.MqConfig;
import com.market.cart.server.mapper.CartInfoMapper;
import com.market.common.core.domain.Result;
import com.market.common.security.service.TokenService;
import com.market.order.common.domain.TOrder;
import com.market.order.common.domain.TOrderDetail;
import com.market.product.common.domain.ProductSkuInfo;
import com.market.product.remote.RemoteProductService;
import com.market.remote.RemoteOrderService;
import com.rabbitmq.client.Channel;
import lombok.extern.log4j.Log4j2;
import org.apache.catalina.core.ApplicationContext;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * <p>ClassName：CartConsumer</p >
 * <p>Description：</p >
 * <p>Author：liuchang</p >
 * <p>Date：2024/3/6</p >
 */
@RabbitListener(queues = MqConfig.ORDER_QUEUE)
@Log4j2
@Component
public class CartConsumer {

    @Autowired
    private RedisTemplate<String,String> redisTemplate;

    @Autowired
    private CartInfoMapper cartInfoMapper;
    @Autowired
    private TokenService tokenService;
    @Autowired
    private RemoteProductService remoteProductService;
    @Autowired
    private RemoteOrderService remoteOrderService;
    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;


    private static final Integer ORDER_STATUS=4;//未发货


    @RabbitHandler
    @Transactional(rollbackFor = Exception.class)
    public void cartConsumer(String jsonString, Message message,Channel channel) throws IOException {
        log.info("收到的消息是：{}",jsonString);

        ObjectMapper mapper = new ObjectMapper();
        HashMap hashMap = mapper.readValue(jsonString, HashMap.class);

        Object object = hashMap.get("userid");
        Long userId = Long.valueOf(object + "");

        if (userId==null){
            return;
        }
        Object objectIds = hashMap.get("ids");
        String strIds = String.valueOf(objectIds);

        Integer[] ids = Arrays.stream(strIds.split(","))
                .map(Integer::parseInt)
                .toArray(Integer[]::new);

       //获取的 MessageID 可以用来判断是否已经被消费者消费过了，如果已经消费则取消再次消费
        String messageId = message.getMessageProperties().getMessageId();
        //它是由 RabbitMQ 生成的整数值，用于跟踪消息的传递状态
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        redisTemplate.opsForValue().set("retry"+messageId,deliveryTag+"");

        try {
            //返回 0，表示没有元素被添加（因为元素已存在）。返回 1 明确指示此次添加操作是第一次
            Long repeat = redisTemplate.opsForSet().add("repeat" + messageId, messageId);

            if (repeat==1) {

              //线程池计算订单总价格
                CompletableFuture<BigDecimal> totalCompletableFuture = CompletableFuture.supplyAsync(() -> {
                    BigDecimal total=BigDecimal.ZERO;

                    for (Integer id : ids) {

                        //根据id查询skuid
                        CartInfo cartInfo = cartInfoMapper.selectCartInfoByCartInfoId(id);
                        Result result = remoteProductService.queryPriceBySkuId(Long.valueOf(cartInfo.getSkuId()));
                        ProductSkuInfo productSkuInfo = JSONObject.parseObject(JSON.toJSONString(result.getData()), ProductSkuInfo.class);
                        BigDecimal price = productSkuInfo.getPrice();

                        BigDecimal cartInfoNum = cartInfo.getCartInfoNum();
                        //小计
                        BigDecimal subtotal = price.multiply(cartInfoNum);

                        total=total.add(subtotal);
                        System.out.println(total);
                    }
                    return total;
                }, threadPoolExecutor);

                  //添加订单表
                CompletableFuture<Integer> orderIdCompletableFuture = totalCompletableFuture.thenApply(total -> {
                    //将价格和基本数据添加到订单表里面 接受返回值，并有返回值
                    TOrder tOrder = new TOrder();
                    tOrder.setAccount(total);
                    tOrder.setOrderStatus(CartConsumer.ORDER_STATUS);
                    tOrder.setOrderTime(new Date());
                    tOrder.setUserId(userId.intValue());
                    Result add = remoteOrderService.addOrder(tOrder);
                    System.out.println("订单表添加成功");
                    Object data = add.getData();
                    Integer orderId = Integer.parseInt(data.toString());
                    System.out.println(orderId);
                    return orderId;
                });

                  //添加详情表
                CompletableFuture<Void> voidCompletableFuture = orderIdCompletableFuture.thenAcceptAsync(orderId -> {
                    //将数据添加到订单详情表里面
                    for (Integer id : ids) {

                        TOrderDetail tOrderDetail = new TOrderDetail();
                        tOrderDetail.setOrderId(orderId);


                        //购物车里面每个商品价格
                        CartInfo cartInfo = cartInfoMapper.selectCartInfoByCartInfoId(id);
                        Result result = remoteProductService.queryPriceBySkuId(Long.valueOf(cartInfo.getSkuId()));
                        ProductSkuInfo productSkuInfo = JSONObject.parseObject(JSON.toJSONString(result.getData()), ProductSkuInfo.class);
                        BigDecimal price = productSkuInfo.getPrice();

                        BigDecimal cartInfoNum = cartInfo.getCartInfoNum();
                        BigDecimal subtotal = price.multiply(cartInfoNum);

                        tOrderDetail.setProductId(cartInfo.getSkuId());
                        tOrderDetail.setProductQuantity(Integer.valueOf(cartInfoNum.intValue()));

                        tOrderDetail.setSubTotal(subtotal);
                        tOrderDetail.setProductPrice(price);
                        tOrderDetail.setCoupon(cartInfo.getCouponId());
                        tOrderDetail.setUserId(userId.intValue());
                        tOrderDetail.setOrderTime(new Date());
                        tOrderDetail.setDeliveryAddress(cartInfo.getAddressId());
                        //添加详情表
                        remoteOrderService.addDetail(tOrderDetail);
                        System.out.println("添加订单详情表");
                    }
                }, threadPoolExecutor);

                 //线程池阻塞
                  CompletableFuture.allOf(voidCompletableFuture).get();

                  //删除购物车
                    cartInfoMapper.deleteCartInfoByCartInfoIds(ids);
                    System.out.println("购物车删除成功");
                    //消费者告知RabbitMQ服务器，它已经成功处理了与指定 deliveryTag 相对应的消息，同时仅确认这一条消息
              channel.basicAck(deliveryTag,false);

            }else {
                //消费者通知RabbitMQ服务器，不去处理 deliveryTag 的消息，直接从队列中移除该消息。 直接拒绝
                channel.basicReject(deliveryTag,false);
                log.info("消息不能重复消费");
            }

        } catch (Exception e) {
            String s = redisTemplate.opsForValue().get("retry" + messageId);
            long l = Long.parseLong(s);
            if (deliveryTag==(l+2)){
                log.info("消息不能再消费，不能进入消息队列");
                channel.basicNack(deliveryTag,false,false);
            }else {
                log.info("消息继续消费，再次进入消息队列");
                channel.basicNack(deliveryTag,false,true);
            }
        }


    }
}
